package com.itdl.mq.base;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamListener;

import java.util.Map;

/**
 * 抽象消费者监听器实现(点对点模式)
 */
@Slf4j
@SuppressWarnings("all")
public abstract class AbstractStreamConsumerListener implements StreamListener<String, MapRecord<String, String, String>> {

    @Autowired
    protected RedisTemplate<String, Object> redisTemplate;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        log.info("=====>>>消费者收消息: StreamName[{}], 消息内容: {}", message.getStream(), message.getValue());

        // 获取生产者消息
        Map<String, String> msgMap = message.getValue();

        // 消息处理
        handleMessage(message);

        // 使用stream进行消息确认
        StreamOperations<String, String, String> streamOperations = redisTemplate.opsForStream();
        streamOperations.acknowledge(message.getStream(),
                getGroupNameByStreamName(message),
                message.getId());
    }

    // 消息处理交给子类自己处理
    protected abstract void handleMessage(MapRecord<String, String, String> message);


    // 根据streamName找到groupName, 因为我们前缀都一样,只是后缀不一样,替换掉即可
    protected String getGroupNameByStreamName(MapRecord<String, String, String> message){
        // 获取到streamName
        String streamName = message.getStream();

        return streamName.substring(0, streamName.lastIndexOf("Stream")) + "Group";
    }
}
